//
// Created by liuping on 2018/12/19.
//

#pragma once

#include <deque>
#include <mutex>
#include <condition_variable>
#include <functional>

template <typename MsgType>
class DoubleQueue {
    std::condition_variable write_q_cv_;
    std::mutex write_q_mutex_;
    std::deque<MsgType> write_msg_q_;
    std::deque<MsgType> read_msg_q_;

    using ConsumeHandler = std::function<int (const MsgType &msg)>;
    ConsumeHandler consume_msg_handler_;
public:
    DoubleQueue() {}
    DoubleQueue(const DoubleQueue &) = delete;
    DoubleQueue& operator=(const DoubleQueue &) = delete;

    template<typename... Args>
    void AddMsg(Args&&... args) {
        {
            std::lock_guard<std::mutex> guard(write_q_mutex_);
            write_msg_q_.emplace_back(std::forward<Args>(args)...);
        }
        write_q_cv_.notify_one();
    }// 通知前完成解锁，以避免等待线程才被唤醒就阻塞（细节见 notify_one ）

    void SetConsumeMsgHandler(ConsumeHandler && consume_handler) {
        consume_msg_handler_ = std::move(consume_handler);
    }

    void OnProc() {
        auto time_out = WaitMsg();
        if (time_out) {
            return;
        }

        Swap();

        ConsumeMsgAll();
    }

 private:
    // 如果没有消息没5ms被唤醒一次
    bool WaitMsg() {
        std::unique_lock<std::mutex> un_lock(write_q_mutex_);
        return !write_q_cv_.wait_for(un_lock, std::chrono::milliseconds(5),
                [this]() { return !(this->write_msg_q_.empty()); });
    }

    void Swap() {
        std::lock_guard<std::mutex> guard(write_q_mutex_);
        if (read_msg_q_.empty() && !write_msg_q_.empty()) {
            write_msg_q_.swap(read_msg_q_);
        }
    }

    void ConsumeMsgAll() {
        while (!read_msg_q_.empty()) {
            const auto &msg = read_msg_q_.front();
            consume_msg_handler_(msg);
            read_msg_q_.pop_front();
        }
    }
};
